package PrjMQTT

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/gohttp"
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"io/ioutil"
	"net/http"
	"strings"
	"time"

	"github.com/gorilla/websocket"
)

var (
	OnHttpMqttSendFilter func(w http.ResponseWriter, r *http.Request, body *[]byte) bool = nil
	mqtt_record_map      utils.GSyncMapGroup
)

type mqtt_record struct {
	remoteaddr      string
	clt_type        string // http, ws
	clt_recvbytes   int64
	clt_last_recv_T time.Time
	last_topic      string
}

var upgrader = websocket.Upgrader{
	ReadBufferSize:  10240,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

func OnMqttWsRequest(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		PrjHttpUtils.ResponseCodeAndMsg(w, -1, err.Error())
		return
	}
	defer conn.Close()

	subKey := strings.TrimPrefix(r.URL.Path, "/ws")
	if len(subKey) == 0 {
		conn.WriteMessage(websocket.TextMessage, []byte("key is empty!"))
		return
	}

	// 尝试绑定
	prjWsMqtt.CheckSubscribleChannel(subKey)

	str := fmt.Sprintf("[订阅消息]:[url:%s][key:%s]", r.URL.Path, subKey)
	utils.Log.Infof(str)
	var jb utils.JSONBuilder
	jb.AddObject("code", 0, "msg", str)
	conn.WriteMessage(websocket.TextMessage, jb.Bytes())

	conn.SetCloseHandler(func(code int, text string) error {
		if prjMqttSC.UnsubscribeChannel(subKey, conn) {
			str := fmt.Sprintf("[退订成功]:[key:%s]", subKey)
			utils.Log.Infof(str)
		} else {
			str := fmt.Sprintf("[退订失败]:[key:%s]", subKey)
			utils.Log.Infof(str)
		}
		return nil
	})

	prjMqttSC.SubscribeChannel(subKey, conn,
		func(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
			if str, ok := data.(string); ok {
				conn.SetWriteDeadline(time.Now().Add(time.Second * 10))
				conn.WriteMessage(websocket.TextMessage, []byte(str))
				return true
			}

			if buf, ok := data.([]byte); ok {
				conn.SetWriteDeadline(time.Now().Add(time.Second * 10))
				conn.WriteMessage(websocket.TextMessage, buf)
				return true
			}

			return false
		})

	for {
		_, _, err := conn.ReadMessage()
		if err != nil {
			golog.Debugf("read:%s", err)
			break
		}
	}
}

func OnMqttWsDisp(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		PrjHttpUtils.ResponseCodeAndMsg(w, -1, err.Error())
		return
	}
	defer conn.Close()
	subKey := strings.TrimPrefix(r.URL.Path, "/ws-send")
	if len(subKey) == 0 {
		conn.WriteMessage(websocket.TextMessage, []byte("key is empty!"))
		return
	}

	for {
		_, msgBuf, err := conn.ReadMessage()
		if err != nil {
			golog.Debugf("read:%s", err)
			break
		} else {
			prjWsMqtt.PublishMsg(subKey, string(msgBuf))
			conn.WriteMessage(websocket.TextMessage, []byte("推送消息:"+string(msgBuf)))
		}
	}
}

func OnHttpMqttStatus(w http.ResponseWriter, r *http.Request) {
	list := mqtt_record_map.ListEx(nil, nil)
	var b utils.BytesBuilder
	for k, v := range list {
		if k == 0 {
			b.AppendStrsWithSpliter("\t", "addr", "type", "last_recv_t").AppendStr("\r\n")
		}
		itm := v.(*mqtt_record)
		b.AppendStrsWithSpliter("\t", itm.remoteaddr, itm.clt_type, utils.DateTimeStringForShow(itm.clt_last_recv_T)).AppendStr("\r\n")
	}
	w.Write(b.Bytes())
}

/***
 * 通过Http接口发送到mqtt服务器
 */
func OnHttpMqttSend(w http.ResponseWriter, r *http.Request) {

	strIp := PrjHttpUtils.GetHttpClientIp(r)
	statusRec := mqtt_record_map.LoadOrStoreFunc(strIp, func() interface{} {
		return &mqtt_record{
			remoteaddr: strIp,
			clt_type:   "http",
		}
	}, nil).(*mqtt_record)

	subKey := strings.TrimPrefix(r.URL.Path, "/mqtt-send")
	if len(subKey) == 0 {
		PrjHttpUtils.ResponseJSONValues(w, "code", -1, "msg", "Topic is empty")
		return
	}
	statusRec.last_topic = subKey

	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		PrjHttpUtils.ResponseJSONValues(w, "code", -1, "msg", utils.EscapeJsonStr(err.Error()))
		return
	}
	statusRec.clt_recvbytes += int64(len(body))
	statusRec.clt_last_recv_T = time.Now()

	if OnHttpMqttSendFilter != nil {
		if !OnHttpMqttSendFilter(w, r, &body) {
			PrjHttpUtils.ResponseJSONValues(w, "code", -1, "msg", utils.EscapeJsonStr("被过滤(OnHttpMqttSendFilter)"))
			return
		}
	}

	if prjWsMqtt.PublishMsg(subKey, string(body)) {
		PrjHttpUtils.ResponseJSONValues(w, "code", 0, "topic", subKey)
		return
	} else {
		PrjHttpUtils.ResponseJSONValues(w, "code", -1,
			"msg", utils.EscapeJsonStr(fmt.Sprintf("[%s]send fail!", subKey)))
		return
	}
}

var (
	prjWsMqtt *ProjMqttService       = NewProjMqttService()
	prjMqttSC *utils.SubscribeCenter = utils.NewSubscribeCenter()
)

func StartProjWsMqtt(conf *wrapper.SuperValue) {
	if conf == nil {
		utils.Log.Warnf("配置不存在:MQTT-WS组件未启动")
	}
	prjWsMqtt.SetRecvCb(OnWsMqttRecv)
	prjWsMqtt.ReloadConfWithoutSubscribleAndStart(conf)
	prjMqttSC.Id = "MQTT-WS消息订阅中心"
	PrjHttpUtils.AddSubscribleToDebugInfo(prjMqttSC)
	registerWsMqttHandlerFuncs()
}

func OnWsMqttRecv(topic string, data []byte) {
	//utils.Log.Infof("接收到信息:" + string(data));
	prjMqttSC.DispatchData(topic, nil, data, nil)
}

func registerWsMqttHandlerFuncs() {
	gohttp.HandleFunc("/ws/", PrjHttpUtils.WrapHandle(OnMqttWsRequest))
	gohttp.HandleFunc("/ws-send/", PrjHttpUtils.WrapHandle(OnMqttWsDisp))
	gohttp.HandleFunc("/mqtt-send/", PrjHttpUtils.WrapHandle(OnHttpMqttSend))
	gohttp.HandleFunc("/mqtt/status", PrjHttpUtils.WrapHandle(OnHttpMqttStatus))
}
